/*******************************************************************************
 * @file fdt_channel.c
 * @brief  inter process Fast Data Transfer
 *
 *
 * Author: Kevinx
 *
 * Copyright (c) 2015 Kevinx. All Rights Reserved.
 *
 * CONFIDENTIALITY AND LIMITED USE
 *
 * This software, including any software of third parties embodied herein,
 * contains information and concepts which are confidential to Kevinx
 * and such third parties. This software is licensed for use
 * solely in accordance with the terms and conditions of the applicable
 * license agreement with Kevinx or its authorized distributor.
 ************************************************************************************/
#include "fdt_channel.h"
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/select.h>
#include <errno.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include "mbuf.h"

#ifdef FDT_DEBUG

static const char* fdt_dir[2] = {"PTRch RX", "PTRch TX"};
#define RX 0
#define TX 1

static void fdt_dump_binary(uint8_t *ptr, int len)
{

    int i, j;
    int addr;
    j = 0;

    printf("===============================\n");

    if (len == 0) {
        printf("nothing to dump\n");
        return;
    }

    for (addr = 0, i = 0; i < len; i ++) {

        if (j%32 == 0)
            printf("0x%04x: ", addr);
      
        j ++;

        if (j == 16) { 
            printf("%02x  ", *(ptr + i));
        } else if (j == 32) {
            printf("%02x\n", *(ptr + i));
            j = 0;
            addr += 32;
        } else {
            printf("%02x ", *(ptr + i));
        }
    }

    printf("\n\n");
    return;
}

#define fdt_dump_msg(dir, ptr, len, off, base, i, count, dump_bin) do { \
    struct mbuf* __m = dtom(ptr); \
    printf("%s, tid: %d, mid: %d, addr: %p, len: %d, off: 0x%08x, base: 0x%08x [%d/%u]\n", \
           fdt_dir[dir], (int)gettid(), m_hdr(__m)->id, ptr, len, off, base, i, count); \
    if (dump_bin) \
        fdt_dump_binary(ptr, len); \
} while (0)

#endif

static int fifo_open(const char* name, int flags)
{
    int fd = -1;

    if ((mkfifo(name, 0777) == -1) && (errno != EEXIST)) {   
        printf("mkfifo: %s: errno=[%d:%s]\n", name, errno, strerror(errno));
        return -1;
    }

    if ((fd = open(name, O_CREAT|flags)) == -1) {
        printf("open fifo: %s: errno=[%d:%s]\n", name, errno, strerror(errno));
        return -1;
    }

    return fd;
}

static int fifo_send(int fd, const char* buf, int size, int no_restart)
{
    int nbytes = -1;
          
    do {
        nbytes = write(fd, (void*)buf, size);
    } while (!no_restart && nbytes < 0 && (errno == EINTR || errno == EAGAIN));


    return nbytes;     
}

static int fifo_receive(int fd, char* buf, int max_nread, int no_restart)
{
    int nbytes = -1;

    do {
        nbytes = read(fd, (void*)buf, max_nread);
    } while (!no_restart && nbytes < 0 && (errno == EINTR || errno == EAGAIN));

    return nbytes;    
}

static int open_pair(int fds[], const char* id, const char* fmt1, const char* fmt2)
{
    char buf[64] = {0};

    sprintf(buf, fmt1, id);
    fds[0] = fifo_open(buf, O_RDWR|O_NONBLOCK);
        
    if (fds[0] == -1)
        return -1;

    sprintf(buf, fmt2, id);
    fds[1] = fifo_open(buf, O_RDWR|O_NONBLOCK);

    if (fds[1] == -1) {
        close(fds[0]);
        return -1;
    }
   
    return 0;    
}

static void close_pair(int fds[])
{
    if (fds[0] != -1)
        close(fds[0]);

    if (fds[1] != -1)
        close(fds[1]);
}

#if 0
static struct shm_area_struct* find_shmarea(struct fdt_channel* c, uint32_t addr)
{
    struct shm_area_struct* shma = NULL;

    if (c) {
        shma = c->shma_cache;
        if (!(shma && shma->end_addr > addr && shma->start_addr <= addr)) {
             
             shma = NULL;
             struct shm_area_struct* shma_tmp;

             list_for_each_entry(shma_tmp, &c->shma_list, list) {
                 if (shma_tmp->end_addr > addr &&
                     shma_tmp->start_addr <= addr) {
                     shma = shma_tmp;
                     break;
                }
            }
            if (shma)
                c->shma_cache = shma;
        }
    }

    return shma;
}
#endif

static int open_PTR_channel(struct fdt_channel* c, struct cid_t* id)
{
    if (open_pair(c->pfds, id->cid, 
                "/tmp/%s.PTR.primary.data.fifo", 
                "/tmp/%s.PTR.primary.gc.fifo"
                ) == -1)
        return -1;
   
    if (open_pair(c->sfds, id->cid, 
                "/tmp/%s.PTR.secondary.data.fifo", 
                "/tmp/%s.PTR.secondary.gc.fifo"
                ) == -1) {
        close_pair(c->pfds);
        return -1;
    }

    c->ref_shma = fdt_shmmgr_attach_shma(id->peerid, id->size);
    if (c->ref_shma == NULL) {
        close_pair(c->pfds);
        close_pair(c->sfds);
        return -1;
    }

    return 0;    
}

static void apply_channel_open_flags(struct fdt_channel* c)
{
    if (c->flags & FCH_NO_NEED_GR) {
        close(get_gr_fd(c));
        get_gr_fd(c) = -1;
    }

    if (c->flags & FCH_NO_NEED_GC) {
        close(get_gc_fd(c));
        get_gc_fd(c) = -1;
    }    
}

/*
 * 
 *
 */
static int open_primary_PTR_channel(struct fdt_channel* c, struct cid_t* id, int flags)
{    
    if (open_PTR_channel(c, id) == -1) 
        return -1;

    c->flags |= flags;
    c->flags |= FCH_PRIMARY;

    c->ofd = &c->pfds[0];
    c->gc_fd = &c->pfds[1];
  
    c->ifd = &c->sfds[0];
    c->gr_fd = &c->sfds[1];

    apply_channel_open_flags(c);

    return 0;
}

/*
 * 
 *
 */
static int open_secondary_PTR_channel(struct fdt_channel* c, struct cid_t* id, int flags)
{
    if (open_PTR_channel(c, id) == -1) 
        return -1;
   
    c->flags |= flags;

    c->ifd = &c->pfds[0];
    c->gr_fd = &c->pfds[1];

    c->ofd = &c->sfds[0];
    c->gc_fd = &c->sfds[1];

    apply_channel_open_flags(c);

    return 0;
}

/*
 * 
 *
 */
static int open_MSG_channel(struct fdt_channel* c, struct cid_t* id)
{
     if (open_pair(c->pfds, id->cid, 
                "/tmp/%s.MSG.data0.fifo", 
                "/tmp/%s.MSG.data1.fifo"
                ) == -1)
        return -1;
   
    return 0;
}

/*
 * 
 *
 */
static int open_primary_MSG_channel(struct fdt_channel* c, struct cid_t* id, int flags)
{
    if (open_MSG_channel(c, id) == -1)
        return -1;
   
    c->flags |= flags;
    c->flags |= FCH_PRIMARY;

    c->ofd = &c->pfds[0];   
    c->ifd = &c->pfds[1];

    return 0;
}

/*
 * 
 *
 */
static int open_secondary_MSG_channel(struct fdt_channel* c, struct cid_t* id, int flags)
{
    if (open_MSG_channel(c, id) == -1)
        return -1;

    c->flags |= flags;

    c->ifd = &c->pfds[0];    
    c->ofd = &c->pfds[1];

    c->gc_fd = NULL;
    c->gr_fd = NULL;

    return 0;
}

/*
 * 
 *
 */
static void close_PTR_channel(struct fdt_channel* c)
{
    close_pair(c->pfds);
    close_pair(c->sfds);

    /*if (c->list.next != NULL)
        list_del(&c->list);*/

    if (c->ref_shma != NULL)
        fdt_shmmgr_del_shma(c->ref_shma);
}

/*
 * 
 *
 */
static void close_MSG_channel(struct fdt_channel* c)
{
    close_pair(c->pfds);

    if (c->list.next != NULL)
        list_del(&c->list);
    
}

static inline int is_primary_channel(struct fdt_channel* c)
{
    return (c->flags & FCH_PRIMARY);
}

static inline int is_secondary_channel(struct fdt_channel* c)
{
    return !(c->flags & FCH_PRIMARY);
}

static int send_PTR_data(struct fdt_channel* c, const void* buf, int size)
{
    struct fdt_buffer* pointer = (struct fdt_buffer*)buf;
    struct shm_area_struct* shma = c->shmmgr->mbuf_area;
#ifdef FDT_DEBUG    
    uint8_t* ptr;
    int len;
#endif

    int i; 
    for (i = 0; i < (int)BYTES_TO_COUNT(size); i++) {

#ifdef FDT_DEBUG      
        if (is_primary_channel(c)) {
            ptr = pointer[i].packet_ptr.ptr;
            len = pointer[i].len;
            fdt_dump_msg(TX, ptr, len, 
                         (pointer[i].packet_ptr.addr - shma->start_addr), 
                         shma->start_addr, i, BYTES_TO_COUNT(size), 0);
        }
#endif
        pointer[i].packet_ptr.addr -= shma->start_addr;
    }
    
    return fifo_send(get_egress_fd(c), buf, size, (c->flags & FCH_NO_RESTART));
}

static int receive_PTR_data(struct fdt_channel* c, void* buf, int max)
{
    int nbytes;
    struct mbuf* m;
    struct shm_area_struct* shma;

    nbytes = fifo_receive(get_ingress_fd(c), buf, max, (c->flags & FCH_NO_RESTART));
    if (nbytes < 0)
        return -1;

    struct fdt_buffer* pointer = (struct fdt_buffer*)buf;
    
    int i; 
    shma = c->ref_shma;
    for (i = 0; i < (int)BYTES_TO_COUNT(nbytes); i++) {
        pointer[i].packet_ptr.addr += shma->start_addr;
#ifdef FDT_DEBUG
        if (is_secondary_channel(c)) {
            fdt_dump_msg(RX,  pointer[i].packet_ptr.ptr, 
                              pointer[i].len, 
                              (pointer[i].packet_ptr.addr - shma->start_addr), 
                              shma->start_addr, i, BYTES_TO_COUNT(nbytes), 0);
        }
#endif        
        m = dtom(pointer[i].packet_ptr.ptr);
        m_hdr(m)->in_ch = c;
    }

    return nbytes;
}

static int send_msg(struct fdt_channel* c, const void* buf, int size)
{
    return fifo_send(get_egress_fd(c), buf, size, (c->flags & FCH_NO_RESTART));
}


static int receive_msg(struct fdt_channel* c, void* buf, int max)
{
    return fifo_receive(get_ingress_fd(c), buf, max, (c->flags & FCH_NO_RESTART));
}

static int poll_ingress_data(int fd, struct timeval* tv)
{
    int rc;
    fd_set rfds;

    FD_ZERO(&rfds);
    FD_SET(fd, &rfds);

    do {
        rc = select(fd+1, &rfds, 0, 0, tv); 
    } while (rc < 0 && errno == EINTR);

    if (rc > 0)
       if (likely(FD_ISSET(fd,  &rfds))) {
            return 1;
       } else
            return -1;
    else 
        return rc; /* rc <= 0*/
}

static int channel_poll_data(struct fdt_channel* c, struct timeval* tv)
{
    return poll_ingress_data(get_ingress_fd(c), tv);
}

static struct fdt_channel_operations primary_PTR_ch_ops = {
    .open = open_primary_PTR_channel,
    .close = close_PTR_channel,
    .send = send_PTR_data,
    .receive = receive_PTR_data,
    .poll = channel_poll_data,
};

static struct fdt_channel_operations secondary_PTR_ch_ops = {
    .open = open_secondary_PTR_channel,
    .close = close_PTR_channel,
    .send = send_PTR_data,
    .receive = receive_PTR_data,
    .poll = channel_poll_data, 
};

static struct fdt_channel_operations primary_MSG_ch_ops = {
    .open = open_primary_MSG_channel,
    .close = close_MSG_channel,
    .send = send_msg,
    .receive = receive_msg,
    .poll = channel_poll_data
};

static struct fdt_channel_operations secondary_MSG_ch_ops = {
    .open = open_secondary_MSG_channel,
    .close = close_MSG_channel,
    .send = send_msg,
    .receive = receive_msg,
    .poll = channel_poll_data   
};

static int make_fdt_channel(struct fdt_channel* c, 
                            struct fdt_shmmgr* shmmgr,
                            struct fdt_channel_operations* ops, 
                            int flags)
{
    if (unlikely(!c))
        return -1;

    c->flags = flags;
    c->ops = ops;
    c->cb = NULL;
    c->shmmgr = shmmgr;
    c->list.prev = c->list.next = NULL;
    /*INIT_LIST_HEAD(&c->shma_list);*/

    return 0;    
}

/*
 * make_primary_PTR_channel
 *      make a primary PTR channel. 
 *
 */
int make_primary_PTR_channel(struct fdt_channel* c, struct fdt_shmmgr* shmmgr)
{
    if (unlikely(!shmmgr))
        return -1;

    return make_fdt_channel(c, shmmgr, &primary_PTR_ch_ops, FCH_TYPE_PTR);
}

/*
 * make_secondary_PTR_channel
 *      make a secondary PTR channel. 
 *
 */
int make_secondary_PTR_channel(struct fdt_channel* c, struct fdt_shmmgr* shmmgr)
{
    if (unlikely(!shmmgr))
        return -1;
    
    return make_fdt_channel(c, shmmgr, &secondary_PTR_ch_ops, FCH_TYPE_PTR);
}

/*
 * make_primary_MSG_channel
 *  make a secondary MSG channel. 
 *
 */
int make_primary_MSG_channel(struct fdt_channel* c)
{
    return make_fdt_channel(c, NULL, &primary_MSG_ch_ops, 0);
}

/*
 * make_secondary_MSG_channel
 *  make a secondary MSG channel. 
 *
 */
int make_secondary_MSG_channel(struct fdt_channel* c)
{
    return make_fdt_channel(c, NULL, &secondary_MSG_ch_ops, 0);
}

/*
 * fdt_channel_open:
 *
 * @return: 0 - success, -1 - failed
 */
int fdt_channel_open(struct fdt_channel* c, struct cid_t* id, int flags)
{
    if (c->ops->open(c, id, flags) == -1)
        return -1;

    return 0;
}

/*
 * fdt_channel_close
 *
 */
void fdt_channel_close(struct fdt_channel* c)
{
    if (likely(c))
        c->ops->close(c);
}

/*
 * fdt_channel_send:
 *  for MSG channel, last two args should be the msg content and it's len.
 *  for PTR channel, "buf" arg are the pointer/len pairs refer to shared memory. len gives the number of pair(len/sizeof(BP)).
 *
 *  @return: number of bytes to send - sucess, 0 - nothing sent, -1 - sending failed
 */
int fdt_channel_send(struct fdt_channel* c, const void* buf, int size)
{
    int rc = -1;

    if (likely(c))
        rc = c->ops->send(c, buf, size);

    return rc;
}

/*
 * fdt_channel_receive:
 *  for MSG channel, last two args are the msg content and it's len.
 *  for PTR channel, "buf" arg will be the pointer/len pairs refer to shared memory. return value gives the number of pair received.
 *
 * @return: number of bytes received - success; -1 - receive failed
 */
int fdt_channel_receive(struct fdt_channel* c, void* buf, int size)
{
    if (likely(c)) 
        return c->ops->receive(c, buf, size);

    return -1;
}

/*
 * fdt_channel_poll: 
 *  polling if there is data to read
 *
 * @return: positive - sucess, 0 - time out, -1 - polling failed
 */
int fdt_channel_poll(struct fdt_channel* c, struct timeval* tv)
{
    if (likely(c))
        return c->ops->poll(c, tv);
    
    return -1;
}

static int PTR_garbage_collect(struct fdt_channel* c)
{
    int nbytes;
    struct mbuf* m;
    struct shm_area_struct* shma;
    struct fdt_buffer pointer[100];

    nbytes = fifo_receive(get_gc_fd(c), (void*)&pointer[0], COUNT_TO_BYTES(100), (c->flags & FCH_NO_RESTART));
    if (nbytes < 0)
        return -1;

    int i;
    shma = c->shmmgr->mbuf_area;

    for (i = 0; i < (int)BYTES_TO_COUNT(nbytes); i++) {
        pointer[i].packet_ptr.addr += shma->start_addr;
        m = dtom(pointer[i].packet_ptr.addr);
        
        m_hdr(m)->flags |= MF_GARBAGE;
        fdt_shmmgr_free(c->shmmgr, pointer[i].packet_ptr.ptr);
    }

    return 0;
}

static int PTR_garbage_return(struct fdt_channel* c, const void* buf, int size)
{
    struct mbuf* m;
    struct fdt_buffer* pointer = (struct fdt_buffer*)buf;
    struct shm_area_struct* shma = c->ref_shma;
    
    int i; 
    for (i = 0; i < (int)BYTES_TO_COUNT(size); i++) {
        m = dtom(pointer[i].packet_ptr.addr);

        pointer[i].packet_ptr.addr -= shma->start_addr;
    }
    
    return fifo_send(get_gr_fd(c), buf, size, (c->flags & FCH_NO_RESTART));    
}

/*
 * fdt_PTRch_gcollect
 *  collect garbage for garbage channel
 *
 * @return: 0 - sucess, -1 - failed
 */
int fdt_PTRch_gcollect(struct fdt_channel* c)
{
    int flags = c->flags;

    if (c && (flags & FCH_TYPE_PTR) && !(flags & FCH_NO_NEED_GC)) {
        return PTR_garbage_collect(c);
    }

    return -1;
}

/*
 * fdt_PTRch_greturn:
 *  return garbarg to peer
 *
 */
int fdt_PTRch_greturn(struct fdt_channel* c, const void* buf, int size)
{
    int flags = c->flags;

    if (c && (flags & FCH_TYPE_PTR) && !(flags & FCH_NO_NEED_GR)) {
        return PTR_garbage_return(c, buf, size);
    }

    return -1;
}

/*
 * fdt_PTRch_GC_poll: 
 *  polling if there is garbage in channel to collect
 *
 * @return: positive - sucess, 0 - time out, -1 - polling failed
 */
int fdt_PTRch_GC_poll(struct fdt_channel* c, struct timeval* tv)
{
    int flags = c->flags;

    if (c && (flags & FCH_TYPE_PTR) && !(flags & FCH_NO_NEED_GC))
        return poll_ingress_data(get_gc_fd(c), tv);

    return -1;
}


